package git.soulbgm.controller;

import com.baomidou.mybatisplus.core.toolkit.Sequence;
import git.soulbgm.common.Constants;
import git.soulbgm.common.KafkaOperate;
import git.soulbgm.pojo.DataSegmentContentSummaryInfo;
import git.soulbgm.pojo.DataSegmentSummaryInfo;
import git.soulbgm.pojo.PacketReceiveSequence;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Random;

/**
 * kafka发送控制器
 *
 * @author SoulBGM
 * @date 2024-09-23
 */
@RestController
@RequestMapping("kafka")
public class KafkaSendController {

    private final KafkaOperate kafka;

    public KafkaSendController(KafkaOperate kafka) {
        this.kafka = kafka;
    }

    private static final Sequence SEQUENCE = new Sequence(null);

    @PostMapping("sendPacketReceiveSequence")
    public String sendPacketReceiveSequence(@RequestBody PacketReceiveSequence sequence) {
        sequence.setPacketId(SEQUENCE.nextId());
        sequence.setDataSegmentNumbering(10);
        sequence.setPacketDate(System.currentTimeMillis());
        kafka.dataCache(Constants.TOPIC_PACKET_RECEIVE_SEQUENCE, sequence);
        return "success";
    }

    private static final Random RANDOM = new Random();

    @PostMapping("sendDataSegmentSummaryInfo")
    public String sendDataSegmentSummaryInfo(@RequestBody DataSegmentSummaryInfo info) {
        info.setId(SEQUENCE.nextId());
        info.setPacketId(SEQUENCE.nextId());
        info.setDataSegmentNumbering(10);
        int sendStatus = RANDOM.nextInt(2);
        info.setReceiveTime(System.currentTimeMillis());
        info.setSendTime(sendStatus == 0 ? 0L : info.getReceiveTime() + RANDOM.nextInt(500));
        info.setSendStatus(sendStatus);
        info.setProcessingTime(sendStatus == 0 ? -1 : (int) (info.getSendTime() - info.getReceiveTime()));
        kafka.dataCache(Constants.TOPIC_DATA_SEGMENT_SUMMARY, info);
        return "success";
    }

    @PostMapping("sendDataSegmentContentSummaryInfo")
    public String sendDataSegmentContentSummaryInfo(@RequestBody DataSegmentContentSummaryInfo info) {
        info.setPacketId(SEQUENCE.nextId());
        info.setDataSegmentNumbering(10);
        info.setReceiveTime(System.currentTimeMillis());
        info.setRdbCard(123456L);
        info.setSdbCard(987654L);
        kafka.dataCache(Constants.TOPIC_DATA_SEGMENT_CONTENT_SUMMARY, info);
        return "success";
    }

}
